package PassengerVehicle

import java.io.{BufferedReader, FileReader}
import java.util.Properties

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 计算运力配置
  * 四个纬度：市内、跨市、省内、省外
  */
object FactorsInOut {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.WARN)
    //创建sparkSession
    val spark: SparkSession = SparkSession.builder().appName("FactorsInOut").master("local[4]").config("spark.debug.maxToStringFields", "1000").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    //创建properties对象 设置连接mysql的信息
    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "123456")
    //读取mysql数据
    val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/localsql?useUnicode=true&characterEncoding=UTF-8", "zhjg_jsy_ticket_smz_curmonth", prop)
    val vehicleDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/localsql?useUnicode=true&characterEncoding=UTF-8", "zhjg_zy_vehicle", prop)
    import spark.sql
    mysqlDF.createOrReplaceTempView("passenger")
    vehicleDF.createOrReplaceTempView("vehicle")
    val mysqlDataFrame = sql("select XL,FCRQSJ,SCKCCPHM,DDZ from passenger where SCKCCPHM ='' or SCKCCPHM like '鲁F%' or SCKCCPHM like '鲁Y%'")
    mysqlDataFrame.createOrReplaceTempView("data")
    sql("select SCKCCPHM,XL,FCRQSJ from data GROUP BY XL,SCKCCPHM,FCRQSJ").createOrReplaceTempView("a")
    //sql("select a.XL, a.SCKCCPHM,count(1) as checi,count(1) * (IF(a.SCKCCPHM = '',50,(SELECT IF(count(1) = 0,50,IFNULL(max(SEAT_COUNT),50)) from vehicle where VEH_NO = a.SCKCCPHM and VEH_STATUS = 10 limit 1))) as hdzkl from a group by XL, SCKCCPHM").show()
    sql("select a.XL, a.SCKCCPHM,count(1) as checi from a group by XL, SCKCCPHM").createOrReplaceTempView("b")
    //sql("SELECT b.XL, b.SCKCCPHM,b.checi,IF (b.SCKCCPHM = '',50,(SELECT if( count( 1 ) = 0, 50, IFNULL( max( SEAT_COUNT ), 50 ) ) FROM vehicle WHERE VEH_NO = b.SCKCCPHM  AND VEH_STATUS = 10  LIMIT 1  ) ) AS hdzkl from b").show()
    sql("SELECT b.XL, b.SCKCCPHM,b.checi  from b")

    /*InCityOrOut(mysqlDataFrame, "incity.txt")
    InProvinceOrOut(mysqlDataFrame, "inprovince.txt")*/

    spark.stop()
  }

  def InCityOrOut(s: DataFrame, fileName: String): Unit = {
    val readerInCity = new FileReader("D:/projects/spark/src/main/resource/" + fileName)
    val bufferReader = new BufferedReader(readerInCity)
    val dict = bufferReader.readLine()
    bufferReader.close()
    readerInCity.close()
    // InCity市内，OutCity，
    val result = s.rdd.map(item => {
      val xl = item.getString(0)
      val ddz = item.getString(3)
      if (matchDDZ(xl, ddz, dict)) {
        ("InCity", 1)
      } else {
        ("OutCity", 1)
      }
    }).reduceByKey((x, y) => x + y)

    result.foreach(x => {
      println(x._1 + "," + x._2)
    })
  }

  def InProvinceOrOut(s: DataFrame, fileName: String): Unit = {
    val readerInCity = new FileReader("D:/projects/spark/src/main/resource/" + fileName)
    val bufferReader = new BufferedReader(readerInCity)
    val dict = bufferReader.readLine()
    bufferReader.close()
    readerInCity.close()
    // InProvince省内，OutProvince省外，
    val result = s.rdd.map(item => {
      val xl = item.getString(0)
      val ddz = item.getString(3)
      if (matchDDZ(xl, ddz, dict)) {
        ("InProvince", 1)
      } else {
        ("OutProvince", 1)
      }
    }).reduceByKey((x, y) => x + y)

    result.foreach(x => {
      println(x._1 + "," + x._2)
    })
  }

  def matchDDZ(xl: String, ddz: String, dict: String): Boolean = {
    val reg1 = "[0-9]"
    val reg2 = "（[\\u4e00-\\u9fa5]*）"
    val reg3 = "[(][0-9\\u4e00-\\u9fa5]*[)]"
    val reg4 = "服务站|下道车|机场|国际机场|车站|汽车站|长途车站|长途汽车站|中心站|中心|高速学生|高速|补|路口|总站|南站|北站|东站|西站|汽车"

    val xllen = xl.split("-").length
    if (dict.indexOf(ddz.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      true
    } else if (xllen == 2 && dict.indexOf(xl.split("-")(1).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      true
    } else if (xllen == 1 && dict.indexOf(xl.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      true
    } else if (xllen == 3 && dict.indexOf(xl.split("-")(2).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
      true
    } else {
      false
    }
  }
}
